package com.galeno.service;

import com.galeno.dao.HistoryEventDao;
import com.galeno.dao.ProfileDao;
import com.galeno.impl.HistoryEventStateDaoImpl;
import com.galeno.impl.ProfileESDaoImpl;
import com.galeno.pojo.*;
import com.galeno.utils.*;
import lombok.extern.log4j.Log4j;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;

/**
 * @author galeno
 * @Title:
 * @Description:
 * @date 2021/11/1121:28
 */
@Log4j
public class QueryService {
    private ProfileDao profileDao;
    private HistoryEventDao eventStateDao;
    private HistoryEventDao historyEventDao;


    public void init(ParameterTool parameterTool, ListState<EventStateBean> eventsState) throws Exception {
//        hbaseConn = ConnectionUtils.getHbaseConn();
//        clickHouseConn = ConnectionUtils.getClickHouseConn();
        String profileDaoClazz = parameterTool.getRequired("profile.dao.impl");
        //通过反射创建实例
        profileDao = (ProfileDao) Class.forName(profileDaoClazz).newInstance();
        profileDao.init(parameterTool);

        String historyEventDaoClazz = parameterTool.getRequired("history.event.dao.impl");

        //创建查询状态的DAO（Data Access Object）
        eventStateDao = new HistoryEventStateDaoImpl(eventsState);


        historyEventDao = (HistoryEventDao) Class.forName(historyEventDaoClazz).newInstance();
        historyEventDao.init(parameterTool);
    }

    public void destroy() throws Exception {
        profileDao.close();
        historyEventDao.close();
    }

    public boolean isMatch(LogBean bean, RuleCondition ruleCondition) throws Exception {
        //匹配当前行为数据
        boolean isMatchTriggerEvent = EventComparator.isMatchTriggerEvent(bean, ruleCondition.getTriggerEvent());
        if (!isMatchTriggerEvent) {
            return false;
        }
        //2.匹配用户画像{tag2 = v1, tag20 = v2}
        boolean isMatchProfile = profileDao.isMatchProfile(bean, ruleCondition.getProfileCondition());
        if (!isMatchProfile) {
            return false;
        }


        return true;
    }

    public boolean isMatchHistoryEvent(LogBean bean, List<CombineCondition> combineConditions) throws Exception {
        for (CombineCondition combineCondition : combineConditions) {

            //根据数据中携带的timeStamp计算查询时间边界
            long timeBound = TimeBoundUtil.getTimeBound(bean.getTimeStamp());
            //事先配置的规则的起始时间、结束时间
            long startTime = combineCondition.getStartTime();
            long endTime = combineCondition.getEndTime();
            //实现配置好的正则表达式例如(1)就是行为次数，如果是(.*?1.*?2.*?3)就是行为序列
            String regex = combineCondition.getRegex();
            //实现设置好的次数条件取出来
            Integer minLimit = combineCondition.getMinLimit();
            //根据时间和条件进行查询
            if (startTime >= timeBound) {
                //到状态中查询
                String sequenceStr = eventStateDao.queryEventSequenceStr(bean, combineCondition);
                int count = MatchUtils.matchCount(sequenceStr, regex);
                if (count < minLimit) return false;
            } else if (endTime <= timeBound) {
                //到clickHouse中查询
                String sequenceStr = historyEventDao.queryEventSequenceStr(bean, combineCondition);
                int count = MatchUtils.matchCount(sequenceStr, regex);
                if (count < minLimit) return false;
            } else {
                //（跨时间边界的）先要到状态中查询，如果状态中满足，直接返回
                long ruleStartTime = combineCondition.getStartTime();
                combineCondition.setStartTime(timeBound);
                String sequenceStr1 = eventStateDao.queryEventSequenceStr(bean, combineCondition);
                combineCondition.setStartTime(ruleStartTime);
                int count = MatchUtils.matchCount(sequenceStr1, regex);
                if (count < minLimit) {
                    long ruleEndTime = combineCondition.getEndTime();
                    combineCondition.setEndTime(timeBound);
                    String sequenceStr2 = historyEventDao.queryEventSequenceStr(bean, combineCondition);
                    combineCondition.setEndTime(ruleEndTime);
                    String resSeq = sequenceStr2 + sequenceStr1;
                    int count2 = MatchUtils.matchCount(resSeq, regex);
                    if (count2 < minLimit) return false;
                }
            }

        }
        return true;


    }


}
